package com.gitee.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment,_}
/*
  1. 确定数据类型,在mysql中创建对应的表
  2. 实现RichSinkFunction接口,从写 open, close, invoke方法
  3. 在invoke中实现保存数据的逻辑
 */
object MySQLSink {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 数据准备
    val studentStream: DataStream[Student] = senv.fromElements(Student(0, "jackma", 18))
    val sql = "insert into account(name, money) values(?,?)"
    val url = "jdbc:mysql://localhost:3306/day21_db"
    val user = "root"
    val password = "root"
    studentStream.addSink(new MySinkToMySQL(sql,url,user,password))
    senv.execute()
  }

  //样例类
  case class Student(id: Int, name: String, age: Int)

  class MySinkToMySQL(sql: String, url: String, user: String, password: String) extends RichSinkFunction[Student] {
    var conn: Connection = _
    var ps: PreparedStatement = _

    //执行保存数据逻辑
    override def invoke(value: Student, context: SinkFunction.Context[_]): Unit = {
      ps.setString(1,value.name)
      ps.setInt(2,value.age)
      ps.executeUpdate()
    }

    override def close(): Unit = {
      if (conn != null) conn.close()
      if (ps != null) ps.close()
    }

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection(url, user, password)
      ps = conn.prepareStatement(sql)
    }
  }

}
